package upClient

import (
	"bytes"
	"common"
	rsEnc "common/encryp/rs_coder"
	"common/erpc"
	rpc "common/rpc"
	"common/uuid"
	"context"
	"errors"
	"euplink/center_rpc"
	"euplink/progress"
	"fmt"
	"go.uber.org/zap"
	"go.uber.org/zap/zapcore"
	"io"
	"time"
)

type PieceUploadResult struct {
	Num       int
	Err       error
	PieceSize int
	NodeId    uuid.UUID
	HashValue []byte
	DevId     string
	Dial      string
	UseTime   time.Duration
}

func (s *PieceUploadResult) MarshalLogObject(enc zapcore.ObjectEncoder) error {
	enc.AddString("devId", s.DevId)
	enc.AddString("dail", s.Dial)
	enc.AddString("speed", common.NetworkSpeedString(int64(s.PieceSize), s.UseTime))
	enc.AddDuration("useTime", s.UseTime)
	enc.AddInt("pieceSize", s.PieceSize)
	return nil
}

type client struct {
	index       int
	log         *zap.Logger
	streamId    []byte
	minUpload   int
	dialer      *rpc.TcpDialer
	expireTime  time.Time
	uploadSpeed int64
	progress    *progress.Progress
}

func NewClient(index int, l *zap.Logger, streamId []byte, minUpload int,
	expireDays time.Time, dialer *rpc.TcpDialer,
	progress *progress.Progress) *client {
	return &client{
		index:      index,
		log:        l,
		streamId:   streamId,
		minUpload:  minUpload,
		progress:   progress,
		expireTime: expireDays,
		dialer:     dialer,
	}
}
func (s *client) Upload(ctx context.Context, data *rsEnc.Pieces, nodes []center_rpc.NodeInfo) (results []PieceUploadResult, err error) {
	sharedCount := len(data.Shared)
	var successfulCount, failureCount int
	defer func() {
		s.log.Info("uploadSegment", zap.String("streamId", fmt.Sprintf("%x", string(s.streamId))),
			zap.Int("segmentIdx", s.index),
			zap.Int("nodeCount", len(nodes)),
			zap.Int("sharedCount", sharedCount),
			zap.Int("successfulCount", successfulCount), zap.Int("failureCount", failureCount),
			zap.Error(err))
	}()

	if len(nodes) < s.minUpload {
		err = errors.New("not enough nodes")
		return
	}

	var piecesCancel context.CancelFunc
	ctx, piecesCancel = context.WithCancel(ctx)
	defer piecesCancel()

	ch := make(chan PieceUploadResult)
	for num, oneShared := range data.Shared {
		if num >= len(nodes) {
			sharedCount--
			continue
		}
		go func(num int, data []byte, nodeInfo center_rpc.NodeInfo) {
			ch <- s.uploadOneShared(ctx, num, data, nodeInfo)
		}(num, oneShared, nodes[num])
	}

	//等待结束
	err = errors.New("not enough upload")
	for i := 0; i < sharedCount; i++ {
		pieceResult := <-ch
		if pieceResult.Err == nil {
			successfulCount++
		} else {
			failureCount++
		}
		results = append(results, pieceResult)
		if successfulCount >= s.minUpload {
			err = nil
			piecesCancel()
		}
	}
	//判断上传成功个数
	return results, err
}

func (s *client) uploadOneShared(ctx context.Context, num int, data []byte, nodeInfo center_rpc.NodeInfo) PieceUploadResult {
	pieceName := fmt.Sprintf("%x-%d-%d", string(s.streamId), s.index, num)
	dataLen := len(data)
	var result = PieceUploadResult{
		Num:       num,
		Err:       nil,
		PieceSize: dataLen,
		NodeId:    nodeInfo.NodeId,
		DevId:     nodeInfo.DevId,
	}
	defer func() {
		s.log.Info(fmt.Sprintf("uploadOneShared[%s]", pieceName), zap.Object("result", &result), zap.Error(result.Err))
	}()

	//优先ipv6
	addr := fmt.Sprintf("%s:%d", nodeInfo.Ip, nodeInfo.Port)
	if nodeInfo.Ipv6 != "" {
		addr = fmt.Sprintf("[%s]:%d", nodeInfo.Ipv6, nodeInfo.Port)
	}
	result.Dial = addr

	conn, err := s.dialer.DialAddress(ctx, addr)
	if err != nil {
		result.Err = err
		return result
	}

	_client := erpc.NewDRPCPiecestoreClient(conn)
	defer func() {
		_client.Close()
	}()

	stream, err := _client.Upload(ctx, pieceName, s.expireTime, s.uploadSpeed)
	if err != nil {
		result.Err = err
		return result
	}
	st := time.Now()
	defer func() {
		result.UseTime = time.Now().Sub(st)
	}()
	err = s.putData(ctx, stream, data, result.DevId, pieceName, num)
	if err != nil {
		result.Err = err
		return result
	}
	response, err := stream.CloseAndRecv()
	if err != nil {
		result.Err = err
		return result
	}

	result.HashValue = response.Hash
	return result

}

func (s *client) putData(ctx context.Context, stream erpc.UploadClient, data []byte, devId string, pieceId string, num int) error {

	var offset = 0
	var maximumStep = 1024
	var dataSize = len(data)
	reader := bytes.NewReader(data)
	sendData := make([]byte, maximumStep)
	var lastPercent = 0
	for {
		var done = 0
		n, readErr := reader.Read(sendData)
		if readErr == io.EOF {
			done = 1
		}
		sendErr := stream.Send(&erpc.Chunk{
			Offset: offset,
			Data:   sendData[:n],
			Done:   done,
		})
		if sendErr != nil {
			return sendErr
		}
		offset += n
		if readErr != nil {
			if readErr == io.EOF {
				break
			} else {
				return readErr
			}
		}
		if ctx.Err() != nil {
			return ctx.Err()
		}
		var percent = int(float32(offset) * 100 / float32(dataSize))
		if lastPercent != percent {
			lastPercent = percent
			//fmt.Println(percent)
			if s.progress != nil {
				s.progress.UpdatePiecePercent(num, int32(lastPercent))
			}
		}
	}
	return nil
}
